package com.itheima.behavior.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.KafkaAdmin;

import java.util.HashMap;
import java.util.Map;

/**
 * Kafka Topic 自动创建配置
 * 启动时自动创建所需的 Topic
 */
@Configuration
@Slf4j
public class KafkaTopicConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    /**
     * Kafka 管理客户端
     */
    @Bean
    public KafkaAdmin kafkaAdmin() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        
        KafkaAdmin admin = new KafkaAdmin(configs);
        // 设置为自动创建 Topic
        admin.setAutoCreate(true);
        
        log.info("Kafka Admin 初始化完成: bootstrapServers={}", bootstrapServers);
        return admin;
    }

    /**
     * 用户行为事件流 Topic
     * 分区数: 3（支持并行消费）
     * 副本数: 1（单节点）
     * 保留时间: 7天
     */
    @Bean
    public NewTopic userBehaviorTopic() {
        return TopicBuilder.name(KafkaConfig.Topics.USER_BEHAVIOR)
                .partitions(3)                    // 3个分区，支持并行处理
                .replicas(1)                      // 单副本（单节点 Kafka）
                .config("retention.ms", "604800000")  // 保留7天 (7*24*60*60*1000)
                .config("compression.type", "lz4")     // 压缩类型
                .build();
    }

    /**
     * 图书行为事件流 Topic
     */
    @Bean
    public NewTopic bookBehaviorTopic() {
        return TopicBuilder.name(KafkaConfig.Topics.BOOK_BEHAVIOR)
                .partitions(3)
                .replicas(1)
                .config("retention.ms", "604800000")
                .config("compression.type", "lz4")
                .build();
    }

    /**
     * 排行榜更新事件 Topic
     */
    @Bean
    public NewTopic rankingUpdateTopic() {
        return TopicBuilder.name(KafkaConfig.Topics.RANKING_UPDATE)
                .partitions(3)
                .replicas(1)
                .config("retention.ms", "86400000")    // 保留1天 (24*60*60*1000)
                .config("compression.type", "lz4")
                .build();
    }
}

